Fox's Git Mirrors
Commit c0d45889ebbae797865487313d1e1efc691032f5
Parents : 0a9917c
Author : ccie18643 <ccie18643@gmail.com>
Date : 2021-10-27T17:38:02Z
Working on FPA unit tests
Changes
14 files changed, 474 insertions(+), 76 deletions(-)
Diff
diff --git a/pytcp/protocols/icmp6/fpa.py b/pytcp/protocols/icmp6/fpa.py
index 250fede..10636c6 100755
--- a/pytcp/protocols/icmp6/fpa.py
+++ b/pytcp/protocols/icmp6/fpa.py
@@ -61,7 +61,7 @@ from protocols.icmp6.ps import (
ICMP6_UNREACHABLE__PORT,
ICMP6_UNREACHABLE_LEN,
)
-from protocols.ip6.ps import IP6_NEXT_HEADER_ICMP6
+from protocols.ip6.ps import IP6_NEXT_ICMP6
if TYPE_CHECKING:
from lib.ip6_address import Ip6Address, Ip6Network
@@ -71,12 +71,12 @@ if TYPE_CHECKING:
class Icmp6Assembler:
"""ICMPv6 packet assembler support class"""
- ip6_next = IP6_NEXT_HEADER_ICMP6
+ ip6_next = IP6_NEXT_ICMP6
def __init__(
self,
*,
- type: int,
+ type: int = 128,
code: int = 0,
un_data: bytes | None = None,
ec_id: int | None = None,
diff --git a/pytcp/protocols/icmp6/phrx.py b/pytcp/protocols/icmp6/phrx.py
index fad7754..e8635c9 100755
--- a/pytcp/protocols/icmp6/phrx.py
+++ b/pytcp/protocols/icmp6/phrx.py
@@ -48,7 +48,7 @@ from protocols.icmp6.ps import (
ICMP6_ND_ROUTER_SOLICITATION,
ICMP6_UNREACHABLE,
)
-from protocols.ip6.ps import IP6_HEADER_LEN, IP6_NEXT_HEADER_UDP
+from protocols.ip6.ps import IP6_HEADER_LEN, IP6_NEXT_UDP
from protocols.udp.metadata import UdpMetadata
from protocols.udp.ps import UDP_HEADER_LEN
@@ -180,7 +180,7 @@ def _phrx_icmp6(self, packet_rx: PacketRx) -> None:
# Quick and dirty way to validate received data and pull useful information from it
# TODO - This will not work in case of IPv6 extension headers present
frame = packet_rx.icmp6.un_data
- if len(frame) >= IP6_HEADER_LEN + UDP_HEADER_LEN and frame[0] >> 4 == 6 and frame[6] == IP6_NEXT_HEADER_UDP:
+ if len(frame) >= IP6_HEADER_LEN + UDP_HEADER_LEN and frame[0] >> 4 == 6 and frame[6] == IP6_NEXT_UDP:
# Create UdpMetadata object and try to find matching UDP socket
udp_offset = IP6_HEADER_LEN
packet = UdpMetadata(
diff --git a/pytcp/protocols/ip4/fpa.py b/pytcp/protocols/ip4/fpa.py
index ffb5436..cf0e297 100755
--- a/pytcp/protocols/ip4/fpa.py
+++ b/pytcp/protocols/ip4/fpa.py
@@ -98,9 +98,9 @@ class Ip4Assembler:
self._src: Ip4Address = src
self._dst: Ip4Address = dst
self._options: list[Ip4OptNop | Ip4OptEol] = [] if options is None else options
+ self._proto: int = self._carried_packet.ip4_proto
self._hlen: int = IP4_HEADER_LEN + len(self._raw_options)
self._plen: int = len(self)
- self._proto: int = self._carried_packet.ip4_proto
def __len__(self) -> int:
"""Length of the packet"""
diff --git a/pytcp/protocols/ip6/fpa.py b/pytcp/protocols/ip6/fpa.py
index 5c8c897..05663dd 100755
--- a/pytcp/protocols/ip6/fpa.py
+++ b/pytcp/protocols/ip6/fpa.py
@@ -39,12 +39,12 @@ from lib.ip6_address import Ip6Address
from protocols.ether.ps import ETHER_TYPE_IP6
from protocols.ip6.ps import (
IP6_HEADER_LEN,
- IP6_NEXT_HEADER_EXT_FRAG,
- IP6_NEXT_HEADER_ICMP6,
- IP6_NEXT_HEADER_RAW,
- IP6_NEXT_HEADER_TABLE,
- IP6_NEXT_HEADER_TCP,
- IP6_NEXT_HEADER_UDP,
+ IP6_NEXT_EXT_FRAG,
+ IP6_NEXT_ICMP6,
+ IP6_NEXT_RAW,
+ IP6_NEXT_TABLE,
+ IP6_NEXT_TCP,
+ IP6_NEXT_UDP,
)
from protocols.raw.fpa import RawAssembler
@@ -79,11 +79,11 @@ class Ip6Assembler:
assert 0 <= ecn <= 0x03
assert 0 <= flow <= 0xFFFFFF
assert carried_packet.ip6_next in {
- IP6_NEXT_HEADER_ICMP6,
- IP6_NEXT_HEADER_UDP,
- IP6_NEXT_HEADER_TCP,
- IP6_NEXT_HEADER_EXT_FRAG,
- IP6_NEXT_HEADER_RAW,
+ IP6_NEXT_ICMP6,
+ IP6_NEXT_UDP,
+ IP6_NEXT_TCP,
+ IP6_NEXT_EXT_FRAG,
+ IP6_NEXT_RAW,
}
self._carried_packet: Ip6ExtFragAssembler | Icmp6Assembler | TcpAssembler | UdpAssembler | RawAssembler = carried_packet
@@ -107,7 +107,7 @@ class Ip6Assembler:
"""Packet log string"""
return (
- f"IPv6 {self._src} > {self._dst}, next {self._next} ({IP6_NEXT_HEADER_TABLE.get(self._next, '???')}), flow {self._flow}"
+ f"IPv6 {self._src} > {self._dst}, next {self._next} ({IP6_NEXT_TABLE.get(self._next, '???')}), flow {self._flow}"
+ f", dlen {self._dlen}, hop {self._hop}"
)
@@ -156,7 +156,7 @@ class Ip6Assembler:
frame,
0,
self._ver << 4 | self._dscp >> 4,
- self._dscp << 6 | self._ecn << 4 | ((self._flow & 0b000011110000000000000000) >> 16),
+ self._dscp & 0b00000011 << 6 | self._ecn << 4 | ((self._flow & 0b000011110000000000000000) >> 16),
(self._flow & 0b000000001111111100000000) >> 8,
self._flow & 0b000000000000000011111111,
self._dlen,
diff --git a/pytcp/protocols/ip6/fpp.py b/pytcp/protocols/ip6/fpp.py
index ffefa74..90a1ace 100755
--- a/pytcp/protocols/ip6/fpp.py
+++ b/pytcp/protocols/ip6/fpp.py
@@ -36,7 +36,7 @@ from typing import TYPE_CHECKING
import config
from lib.ip6_address import Ip6Address
-from protocols.ip6.ps import IP6_HEADER_LEN, IP6_NEXT_HEADER_TABLE
+from protocols.ip6.ps import IP6_HEADER_LEN, IP6_NEXT_TABLE
if TYPE_CHECKING:
from misc.packet import PacketRx
@@ -67,8 +67,7 @@ class Ip6Parser:
"""Packet log string"""
return (
- f"IPv6 {self.src} > {self.dst}, next {self.next} ({IP6_NEXT_HEADER_TABLE.get(self.next, '???')}), flow {self.flow}"
- + f", dlen {self.dlen}, hop {self.hop}"
+ f"IPv6 {self.src} > {self.dst}, next {self.next} ({IP6_NEXT_TABLE.get(self.next, '???')}), flow {self.flow}" + f", dlen {self.dlen}, hop {self.hop}"
)
@property
diff --git a/pytcp/protocols/ip6/phrx.py b/pytcp/protocols/ip6/phrx.py
index 99fceee..0e33bca 100755
--- a/pytcp/protocols/ip6/phrx.py
+++ b/pytcp/protocols/ip6/phrx.py
@@ -35,10 +35,10 @@ from lib.logger import log
from misc.packet import PacketRx
from protocols.ip6.fpp import Ip6Parser
from protocols.ip6.ps import (
- IP6_NEXT_HEADER_EXT_FRAG,
- IP6_NEXT_HEADER_ICMP6,
- IP6_NEXT_HEADER_TCP,
- IP6_NEXT_HEADER_UDP,
+ IP6_NEXT_EXT_FRAG,
+ IP6_NEXT_ICMP6,
+ IP6_NEXT_TCP,
+ IP6_NEXT_UDP,
)
@@ -71,18 +71,18 @@ def _phrx_ip6(self, packet_rx: PacketRx) -> None:
if packet_rx.ip6.dst in self.ip6_multicast:
self.packet_stats_rx.ip6__dst_multicast += 1
- if packet_rx.ip6.next == IP6_NEXT_HEADER_EXT_FRAG:
+ if packet_rx.ip6.next == IP6_NEXT_EXT_FRAG:
self._phrx_ip6_ext_frag(packet_rx)
return
- if packet_rx.ip6.next == IP6_NEXT_HEADER_ICMP6:
+ if packet_rx.ip6.next == IP6_NEXT_ICMP6:
self._phrx_icmp6(packet_rx)
return
- if packet_rx.ip6.next == IP6_NEXT_HEADER_UDP:
+ if packet_rx.ip6.next == IP6_NEXT_UDP:
self._phrx_udp(packet_rx)
return
- if packet_rx.ip6.next == IP6_NEXT_HEADER_TCP:
+ if packet_rx.ip6.next == IP6_NEXT_TCP:
self._phrx_tcp(packet_rx)
return
diff --git a/pytcp/protocols/ip6/ps.py b/pytcp/protocols/ip6/ps.py
index 8b3e8d7..6008bfc 100755
--- a/pytcp/protocols/ip6/ps.py
+++ b/pytcp/protocols/ip6/ps.py
@@ -58,16 +58,16 @@ from __future__ import annotations
IP6_HEADER_LEN = 40
-IP6_NEXT_HEADER_TCP = 6
-IP6_NEXT_HEADER_UDP = 17
-IP6_NEXT_HEADER_EXT_FRAG = 44
-IP6_NEXT_HEADER_ICMP6 = 58
-IP6_NEXT_HEADER_RAW = 255
+IP6_NEXT_TCP = 6
+IP6_NEXT_UDP = 17
+IP6_NEXT_EXT_FRAG = 44
+IP6_NEXT_ICMP6 = 58
+IP6_NEXT_RAW = 255
-IP6_NEXT_HEADER_TABLE = {
- IP6_NEXT_HEADER_TCP: "TCP",
- IP6_NEXT_HEADER_UDP: "UDP",
- IP6_NEXT_HEADER_EXT_FRAG: "FRAG",
- IP6_NEXT_HEADER_ICMP6: "ICMPv6",
- IP6_NEXT_HEADER_RAW: "raw_data",
+IP6_NEXT_TABLE = {
+ IP6_NEXT_TCP: "TCP",
+ IP6_NEXT_UDP: "UDP",
+ IP6_NEXT_EXT_FRAG: "FRAG",
+ IP6_NEXT_ICMP6: "ICMPv6",
+ IP6_NEXT_RAW: "raw_data",
}
diff --git a/pytcp/protocols/ip6_ext_frag/fpa.py b/pytcp/protocols/ip6_ext_frag/fpa.py
index 45923c1..cd13c07 100755
--- a/pytcp/protocols/ip6_ext_frag/fpa.py
+++ b/pytcp/protocols/ip6_ext_frag/fpa.py
@@ -35,11 +35,11 @@ import struct
from lib.tracker import Tracker
from protocols.ip6.ps import (
- IP6_NEXT_HEADER_EXT_FRAG,
- IP6_NEXT_HEADER_ICMP6,
- IP6_NEXT_HEADER_RAW,
- IP6_NEXT_HEADER_TCP,
- IP6_NEXT_HEADER_UDP,
+ IP6_NEXT_EXT_FRAG,
+ IP6_NEXT_ICMP6,
+ IP6_NEXT_RAW,
+ IP6_NEXT_TCP,
+ IP6_NEXT_UDP,
)
from protocols.ip6_ext_frag.ps import (
IP6_EXT_FRAG_HEADER_LEN,
@@ -50,7 +50,7 @@ from protocols.ip6_ext_frag.ps import (
class Ip6ExtFragAssembler:
"""IPv6 fragment extension header assembler support class"""
- ip6_next = IP6_NEXT_HEADER_EXT_FRAG
+ ip6_next = IP6_NEXT_EXT_FRAG
def __init__(
self,
@@ -63,7 +63,7 @@ class Ip6ExtFragAssembler:
):
"""Class constructor"""
- assert next in {IP6_NEXT_HEADER_ICMP6, IP6_NEXT_HEADER_UDP, IP6_NEXT_HEADER_TCP, IP6_NEXT_HEADER_RAW}
+ assert next in {IP6_NEXT_ICMP6, IP6_NEXT_UDP, IP6_NEXT_TCP, IP6_NEXT_RAW}
self._tracker: Tracker = Tracker(prefix="TX")
self._next: int = next
diff --git a/pytcp/protocols/raw/fpa.py b/pytcp/protocols/raw/fpa.py
index 80ade2c..c49aecd 100755
--- a/pytcp/protocols/raw/fpa.py
+++ b/pytcp/protocols/raw/fpa.py
@@ -36,14 +36,14 @@ import struct
from lib.tracker import Tracker
from protocols.ether.ps import ETHER_TYPE_RAW
from protocols.ip4.ps import IP4_PROTO_RAW
-from protocols.ip6.ps import IP6_NEXT_HEADER_RAW
+from protocols.ip6.ps import IP6_NEXT_RAW
class RawAssembler:
"""Raw packet assembler support class"""
ip4_proto = IP4_PROTO_RAW
- ip6_next = IP6_NEXT_HEADER_RAW
+ ip6_next = IP6_NEXT_RAW
ether_type = ETHER_TYPE_RAW
def __init__(self, *, data: bytes | None = None, echo_tracker: Tracker | None = None) -> None:
@@ -63,6 +63,16 @@ class RawAssembler:
return f"Raw, len {self._plen}"
+ def __repr__(self) -> str:
+ """Object representation"""
+
+ return f"RawAssembler(data={self._data})"
+
+ def __eq__(self, other) -> bool:
+ """Equal operator"""
+
+ return repr(self) == repr(other)
+
@property
def tracker(self) -> Tracker:
"""Getter for _tracker"""
diff --git a/pytcp/protocols/tcp/fpa.py b/pytcp/protocols/tcp/fpa.py
index a77b69b..613fd75 100755
--- a/pytcp/protocols/tcp/fpa.py
+++ b/pytcp/protocols/tcp/fpa.py
@@ -36,7 +36,7 @@ import struct
from lib.tracker import Tracker
from misc.ip_helper import inet_cksum
from protocols.ip4.ps import IP4_PROTO_TCP
-from protocols.ip6.ps import IP6_NEXT_HEADER_TCP
+from protocols.ip6.ps import IP6_NEXT_TCP
from protocols.tcp.ps import (
TCP_HEADER_LEN,
TCP_OPT_EOL,
@@ -58,7 +58,7 @@ class TcpAssembler:
"""TCP packet assembler support class"""
ip4_proto = IP4_PROTO_TCP
- ip6_next = IP6_NEXT_HEADER_TCP
+ ip6_next = IP6_NEXT_TCP
def __init__(
self,
diff --git a/pytcp/protocols/udp/fpa.py b/pytcp/protocols/udp/fpa.py
index eb0bf13..d0cd2b4 100755
--- a/pytcp/protocols/udp/fpa.py
+++ b/pytcp/protocols/udp/fpa.py
@@ -36,7 +36,7 @@ import struct
from lib.tracker import Tracker
from misc.ip_helper import inet_cksum
from protocols.ip4.ps import IP4_PROTO_UDP
-from protocols.ip6.ps import IP6_NEXT_HEADER_UDP
+from protocols.ip6.ps import IP6_NEXT_UDP
from protocols.udp.ps import UDP_HEADER_LEN
@@ -44,7 +44,7 @@ class UdpAssembler:
"""UDP packet assembler support class"""
ip4_proto = IP4_PROTO_UDP
- ip6_next = IP6_NEXT_HEADER_UDP
+ ip6_next = IP6_NEXT_UDP
def __init__(self, *, sport: int = 0, dport: int = 0, data: bytes | None = None, echo_tracker: Tracker | None = None) -> None:
"""Class constructor"""
diff --git a/tests/ether_fpa.py b/tests/ether_fpa.py
index d41fb1e..4fe6e98 100755
--- a/tests/ether_fpa.py
+++ b/tests/ether_fpa.py
@@ -42,27 +42,27 @@ from pytcp.protocols.raw.fpa import RawAssembler
class TestEtherAssembler(TestCase):
- def test_ether_fpa__ethertype_arp(self):
+ def test_ether_fpa__assert_ethertype_arp(self):
"""Test assertion for carried packet ether_type attribute"""
EtherAssembler(carried_packet=ArpAssembler())
- def test_ether_fpa__ethertype_ip4(self):
+ def test_ether_fpa__assert_ethertype_ip4(self):
"""Test assertion for carried packet ether_type attribute"""
EtherAssembler(carried_packet=Ip4Assembler())
- def test_ether_fpa__ethertype_ip6(self):
+ def test_ether_fpa__assert_ethertype_ip6(self):
"""Test assertion for carried packet ether_type attribute"""
EtherAssembler(carried_packet=Ip6Assembler())
- def test_ether_fpa__ethertype_raw(self):
+ def test_ether_fpa__assert_ethertype_raw(self):
"""Test assertion for carried packet ether_type attribute"""
EtherAssembler(carried_packet=RawAssembler())
- def test_ether_fpa__ethertype_unknown(self):
+ def test_ether_fpa__assert_ethertype_unknown(self):
"""Test assertion for carried packet ether_type attribute"""
with self.assertRaises(AssertionError):
@@ -80,6 +80,8 @@ class TestEtherAssembler(TestCase):
carried_packet=RawAssembler(),
)
+ self.assertEqual(packet._carried_packet, RawAssembler())
+ self.assertEqual(packet._tracker, packet._carried_packet._tracker)
self.assertEqual(packet._src, MacAddress("00:11:22:33:44:55"))
self.assertEqual(packet._dst, MacAddress("66:77:88:99:AA:BB"))
self.assertEqual(packet._type, ETHER_TYPE_RAW)
@@ -89,6 +91,8 @@ class TestEtherAssembler(TestCase):
packet = EtherAssembler()
+ self.assertEqual(packet._carried_packet, RawAssembler())
+ self.assertEqual(packet._tracker, packet._carried_packet._tracker)
self.assertEqual(packet._src, MacAddress("00:00:00:00:00:00"))
self.assertEqual(packet._dst, MacAddress("00:00:00:00:00:00"))
self.assertEqual(packet._type, ETHER_TYPE_RAW)
diff --git a/tests/ip4_fpa.py b/tests/ip4_fpa.py
index 9925ebc..8ff80d5 100755
--- a/tests/ip4_fpa.py
+++ b/tests/ip4_fpa.py
@@ -31,6 +31,7 @@
from testslide import StrictMock, TestCase
+from pytcp.config import IP4_DEFAULT_TTL
from pytcp.lib.ip4_address import Ip4Address
from pytcp.lib.tracker import Tracker
from pytcp.protocols.ether.ps import ETHER_TYPE_IP4
@@ -40,10 +41,10 @@ from pytcp.protocols.ip4.ps import (
IP4_HEADER_LEN,
IP4_OPT_EOL_LEN,
IP4_OPT_NOP_LEN,
- IP4_PROTO_TCP,
- IP4_PROTO_UDP,
IP4_PROTO_ICMP4,
IP4_PROTO_RAW,
+ IP4_PROTO_TCP,
+ IP4_PROTO_UDP,
)
from pytcp.protocols.raw.fpa import RawAssembler
from pytcp.protocols.tcp.fpa import TcpAssembler
@@ -164,7 +165,7 @@ class TestIp4Assembler(TestCase):
Ip4Assembler(carried_packet=RawAssembler())
- def test_ether_fpa__proto_unknown(self):
+ def test_ip4_fpa__assert_proto_unknown(self):
"""Test assertion for carried packet p4_proto attribute"""
with self.assertRaises(AssertionError):
@@ -185,35 +186,47 @@ class TestIp4Assembler(TestCase):
id=12345,
flag_df=True,
options=[Ip4OptNop(), Ip4OptEol()],
- carried_packet=RawAssembler(),
+ carried_packet=RawAssembler(data=b"0123456789ABCDEF"),
)
- self.assertEqual(packet._src, Ip4Address("1.2.3.4"))
- self.assertEqual(packet._dst, Ip4Address("5.6.7.8"))
- self.assertEqual(packet._ttl, 32)
+ self.assertEqual(packet._carried_packet, RawAssembler(data=b"0123456789ABCDEF"))
+ self.assertEqual(packet._tracker, packet._carried_packet._tracker)
+ self.assertEqual(packet._ver, 4)
self.assertEqual(packet._dscp, 10)
self.assertEqual(packet._ecn, 2)
self.assertEqual(packet._id, 12345)
self.assertEqual(packet._flag_df, True)
+ self.assertEqual(packet._flag_mf, False)
self.assertEqual(packet._offset, 0)
+ self.assertEqual(packet._ttl, 32)
+ self.assertEqual(packet._src, Ip4Address("1.2.3.4"))
+ self.assertEqual(packet._dst, Ip4Address("5.6.7.8"))
self.assertEqual(packet._options, [Ip4OptNop(), Ip4OptEol()])
self.assertEqual(packet._proto, IP4_PROTO_RAW)
+ self.assertEqual(packet._hlen, IP4_HEADER_LEN + IP4_OPT_NOP_LEN + IP4_OPT_EOL_LEN)
+ self.assertEqual(packet._plen, IP4_HEADER_LEN + IP4_OPT_NOP_LEN + IP4_OPT_EOL_LEN + 16)
def test_ip4_fpa__constructor__defaults(self):
"""Test class constructor"""
packet = Ip4Assembler()
- self.assertEqual(packet._src, Ip4Address(0))
- self.assertEqual(packet._dst, Ip4Address(0))
- self.assertEqual(packet._ttl, 64)
+ self.assertEqual(packet._carried_packet, RawAssembler())
+ self.assertEqual(packet._tracker, packet._carried_packet._tracker)
+ self.assertEqual(packet._ver, 4)
self.assertEqual(packet._dscp, 0)
self.assertEqual(packet._ecn, 0)
self.assertEqual(packet._id, 0)
self.assertEqual(packet._flag_df, False)
+ self.assertEqual(packet._flag_mf, False)
self.assertEqual(packet._offset, 0)
+ self.assertEqual(packet._ttl, IP4_DEFAULT_TTL)
+ self.assertEqual(packet._src, Ip4Address(0))
+ self.assertEqual(packet._dst, Ip4Address(0))
self.assertEqual(packet._options, [])
self.assertEqual(packet._proto, IP4_PROTO_RAW)
+ self.assertEqual(packet._hlen, IP4_HEADER_LEN)
+ self.assertEqual(packet._plen, IP4_HEADER_LEN)
def test_ip4_fpa__len(self):
"""Test class __len__ operator"""
@@ -342,9 +355,13 @@ class TestIp4Assembler(TestCase):
def test_ip4_fpa__pshdr_sum(self):
"""Test pshdr_sum getter"""
- packet = Ip4Assembler()
+ packet = Ip4Assembler(
+ src=Ip4Address("1.2.3.4"),
+ dst=Ip4Address("5.6.7.8"),
+ carried_packet=RawAssembler(data=b"0123456789ABCDEF"),
+ )
- self.assertEqual(packet.pshdr_sum, 16711680)
+ self.assertEqual(packet.pshdr_sum, 117901852)
def test_ip4_fpa___raw_options(self):
"""Test _raw_options getter"""
@@ -376,7 +393,7 @@ class TestIp4Assembler(TestCase):
class TestIp4FragAssembler(TestCase):
- def test_ip4frag_fpa__ethertype(self):
+ def test_ip4_frag_fpa__ethertype(self):
"""Test the ethertype of Ip4Assembler class"""
def test_ip4_frag_fpa__assert_ttl(self):
@@ -487,12 +504,65 @@ class TestIp4FragAssembler(TestCase):
Ip4FragAssembler(proto=IP4_PROTO_RAW)
- def test_ip4_frag_fpa__proto_unknown(self):
+ def test_ip4_frag_fpa__assert_proto_unknown(self):
"""Test assertion for carried packet p4_proto attribute"""
with self.assertRaises(AssertionError):
Ip4FragAssembler(proto=-1)
+ def test_ip4_frag_fpa__constructor(self):
+ """Test class constructor"""
+
+ packet = Ip4FragAssembler(
+ src=Ip4Address("1.2.3.4"),
+ dst=Ip4Address("5.6.7.8"),
+ ttl=32,
+ dscp=10,
+ ecn=2,
+ id=12345,
+ flag_mf=True,
+ options=[Ip4OptNop(), Ip4OptEol()],
+ data=b"0123456789ABCDEF",
+ proto=IP4_PROTO_RAW,
+ )
+
+ self.assertEqual(packet._ver, 4)
+ self.assertEqual(packet._dscp, 10)
+ self.assertEqual(packet._ecn, 2)
+ self.assertEqual(packet._id, 12345)
+ self.assertEqual(packet._flag_df, False)
+ self.assertEqual(packet._flag_mf, True)
+ self.assertEqual(packet._offset, 0)
+ self.assertEqual(packet._ttl, 32)
+ self.assertEqual(packet._src, Ip4Address("1.2.3.4"))
+ self.assertEqual(packet._dst, Ip4Address("5.6.7.8"))
+ self.assertEqual(packet._options, [Ip4OptNop(), Ip4OptEol()])
+ self.assertEqual(packet._data, b"0123456789ABCDEF")
+ self.assertEqual(packet._proto, IP4_PROTO_RAW)
+ self.assertEqual(packet._hlen, IP4_HEADER_LEN + IP4_OPT_NOP_LEN + IP4_OPT_EOL_LEN)
+ self.assertEqual(packet._plen, IP4_HEADER_LEN + IP4_OPT_NOP_LEN + IP4_OPT_EOL_LEN + 16)
+
+ def test_ip4_frag_fpa__constructor__defaults(self):
+ """Test class constructor"""
+
+ packet = Ip4FragAssembler()
+
+ self.assertEqual(packet._ver, 4)
+ self.assertEqual(packet._dscp, 0)
+ self.assertEqual(packet._ecn, 0)
+ self.assertEqual(packet._id, 0)
+ self.assertEqual(packet._flag_df, False)
+ self.assertEqual(packet._flag_mf, False)
+ self.assertEqual(packet._offset, 0)
+ self.assertEqual(packet._ttl, IP4_DEFAULT_TTL)
+ self.assertEqual(packet._src, Ip4Address(0))
+ self.assertEqual(packet._dst, Ip4Address(0))
+ self.assertEqual(packet._options, [])
+ self.assertEqual(packet._data, b"")
+ self.assertEqual(packet._proto, IP4_PROTO_RAW)
+ self.assertEqual(packet._hlen, IP4_HEADER_LEN)
+ self.assertEqual(packet._plen, IP4_HEADER_LEN)
+
def test_ip4_frag_fpa__len(self):
"""Test class __len__ operator"""
@@ -528,7 +598,7 @@ class TestIp4FragAssembler(TestCase):
)
self.assertEqual(len(packet), IP4_HEADER_LEN + IP4_OPT_NOP_LEN + IP4_OPT_EOL_LEN + 16)
-
+
def test_ip4_frag_fpa__str__(self):
"""Test class __str__ operator"""
@@ -543,7 +613,7 @@ class TestIp4FragAssembler(TestCase):
offset=54321,
options=[],
proto=IP4_PROTO_RAW,
- data=b""
+ data=b"",
)
self.assertEqual(str(packet), "IPv4 1.2.3.4 > 5.6.7.8, proto 255 (raw_data), id 12345, MF, offset 54321, plen 20, ttl 32")
@@ -562,7 +632,7 @@ class TestIp4FragAssembler(TestCase):
offset=54321,
options=[Ip4OptNop(), Ip4OptEol()],
proto=IP4_PROTO_RAW,
- data=b""
+ data=b"",
)
self.assertEqual(str(packet), "IPv4 1.2.3.4 > 5.6.7.8, proto 255 (raw_data), id 12345, MF, offset 54321, plen 22, ttl 32, nop, eol")
diff --git a/tests/ip6_fpa.py b/tests/ip6_fpa.py
new file mode 100755
index 0000000..cad4cec
--- /dev/null
+++ b/tests/ip6_fpa.py
@@ -0,0 +1,315 @@
+#!/usr/bin/env python3
+
+
+############################################################################
+# #
+# PyTCP - Python TCP/IP stack #
+# Copyright (C) 2020-2021 Sebastian Majewski #
+# #
+# This program is free software: you can redistribute it and/or modify #
+# it under the terms of the GNU General Public License as published by #
+# the Free Software Foundation, either version 3 of the License, or #
+# (at your option) any later version. #
+# #
+# This program is distributed in the hope that it will be useful, #
+# but WITHOUT ANY WARRANTY; without even the implied warranty of #
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
+# GNU General Public License for more details. #
+# #
+# You should have received a copy of the GNU General Public License #
+# along with this program. If not, see <https://www.gnu.org/licenses/>. #
+# #
+# Author's email: ccie18643@gmail.com #
+# Github repository: https://github.com/ccie18643/PyTCP #
+# #
+############################################################################
+
+
+#
+# tests/ip6_fpa.py - tests specific for IPv6 fpa module
+#
+
+from testslide import StrictMock, TestCase
+
+from pytcp.config import IP6_DEFAULT_HOP
+from pytcp.lib.ip6_address import Ip6Address
+from pytcp.lib.tracker import Tracker
+from pytcp.protocols.ether.ps import ETHER_TYPE_IP6
+from pytcp.protocols.icmp6.fpa import Icmp6Assembler
+from pytcp.protocols.ip6.fpa import Ip6Assembler
+from pytcp.protocols.ip6.ps import (
+ IP6_HEADER_LEN,
+ IP6_NEXT_ICMP6,
+ IP6_NEXT_RAW,
+ IP6_NEXT_TCP,
+ IP6_NEXT_UDP,
+)
+from pytcp.protocols.raw.fpa import RawAssembler
+from pytcp.protocols.tcp.fpa import TcpAssembler
+from pytcp.protocols.udp.fpa import UdpAssembler
+
+
+class TestIp6Assembler(TestCase):
+ def test_ip6_fpa__ethertype(self):
+ """Test the ethertype of Ip6Assembler class"""
+
+ self.assertEqual(Ip6Assembler.ether_type, ETHER_TYPE_IP6)
+
+ def test_ip6_fpa__assert_hop(self):
+ """Test assertion for the hop"""
+
+ Ip6Assembler(hop=0x20)
+
+ def test_ip6_fpa__assert_hop__default(self):
+ """Test assertion for the hop"""
+
+ Ip6Assembler()
+
+ def test_ip6_fpa__assert_hop__bellow(self):
+ """Test assertion for the hop"""
+
+ with self.assertRaises(AssertionError):
+ Ip6Assembler(hop=-1)
+
+ def test_ip6_fpa__assert_hop__above(self):
+ """Test assertion for the hop"""
+
+ with self.assertRaises(AssertionError):
+ Ip6Assembler(hop=0x100)
+
+ def test_ip6_fpa__assert_dscp(self):
+ """Test assertion for the dscp"""
+
+ Ip6Assembler(dscp=0x10)
+
+ def test_ip6_fpa__assert_dscp__default(self):
+ """Test assertion for the dscp"""
+
+ Ip6Assembler()
+
+ def test_ip6_fpa__assert_dscp__bellow(self):
+ """Test assertion for the dscp"""
+
+ with self.assertRaises(AssertionError):
+ Ip6Assembler(dscp=-1)
+
+ def test_ip6_fpa__assert_dscp__above(self):
+ """Test assertion for the dscp"""
+
+ with self.assertRaises(AssertionError):
+ Ip6Assembler(dscp=0x40)
+
+ def test_ip6_fpa__assert_ecn(self):
+ """Test assertion for the ecn"""
+
+ Ip6Assembler(ecn=2)
+
+ def test_ip6_fpa__assert_ecn__default(self):
+ """Test assertion for the ecn"""
+
+ Ip6Assembler()
+
+ def test_ip6_fpa__assert_ecn__bellow(self):
+ """Test assertion for the ecn"""
+
+ with self.assertRaises(AssertionError):
+ Ip6Assembler(ecn=-1)
+
+ def test_ip6_fpa__assert_ecn__above(self):
+ """Test assertion for the ecn"""
+
+ with self.assertRaises(AssertionError):
+ Ip6Assembler(ecn=4)
+
+ def test_ip6_fpa__assert_flow(self):
+ """Test assertion for the flow"""
+
+ Ip6Assembler(flow=12345)
+
+ def test_ip6_fpa__assert_flow__default(self):
+ """Test assertion for the flow"""
+
+ Ip6Assembler()
+
+ def test_ip6_fpa__assert_flow__bellow(self):
+ """Test assertion for the flow"""
+
+ with self.assertRaises(AssertionError):
+ Ip6Assembler(flow=-1)
+
+ def test_ip6_fpa__assert_flow__above(self):
+ """Test assertion for the flow"""
+
+ with self.assertRaises(AssertionError):
+ Ip6Assembler(flow=0x1000000)
+
+ def test_ip6_fpa__assert_next_header_udp(self):
+ """Test assertion for carried packet ip6_next_header attribute"""
+
+ Ip6Assembler(carried_packet=UdpAssembler())
+
+ def test_ip6_fpa__assert_next_header_tcp(self):
+ """Test assertion for carried packet ip6_next_header attribute"""
+
+ Ip6Assembler(carried_packet=TcpAssembler())
+
+ def test_ip6_fpa__assert_next_header_icmp6(self):
+ """Test assertion for carried packet ip6_next_header attribute"""
+
+ Ip6Assembler(carried_packet=Icmp6Assembler())
+
+ def test_ip6_fpa__assert_next_header_raw(self):
+ """Test assertion for carried packet ip6_next_header attribute"""
+
+ Ip6Assembler(carried_packet=RawAssembler())
+
+ def test_ether_fpa__next_header_unknown(self):
+ """Test assertion for carried packet p4_next_header attribute"""
+
+ with self.assertRaises(AssertionError):
+ carried_packet_mock = StrictMock()
+ carried_packet_mock.ip6_next = -1
+ carried_packet_mock.tracker = StrictMock(Tracker)
+ Ip6Assembler(carried_packet=carried_packet_mock)
+
+ def test_ip6_fpa__constructor(self):
+ """Test class constructor"""
+
+ packet = Ip6Assembler(
+ src=Ip6Address("0:1:2:3:4:5:6:7"),
+ dst=Ip6Address("8:9:A:B:C:D:E:F"),
+ hop=32,
+ dscp=10,
+ ecn=2,
+ flow=12345678,
+ carried_packet=RawAssembler(data=b"0123456789ABCDEF"),
+ )
+
+ self.assertEqual(packet._carried_packet, RawAssembler(data=b"0123456789ABCDEF"))
+ self.assertEqual(packet._tracker, packet._carried_packet._tracker)
+ self.assertEqual(packet._ver, 6)
+ self.assertEqual(packet._dscp, 10)
+ self.assertEqual(packet._ecn, 2)
+ self.assertEqual(packet._flow, 12345678)
+ self.assertEqual(packet._hop, 32)
+ self.assertEqual(packet._src, Ip6Address("0:1:2:3:4:5:6:7"))
+ self.assertEqual(packet._dst, Ip6Address("8:9:A:B:C:D:E:F"))
+ self.assertEqual(packet._next, IP6_NEXT_RAW)
+ self.assertEqual(packet._dlen, 16)
+
+ def test_ip6_fpa__constructor__defaults(self):
+ """Test class constructor"""
+
+ packet = Ip6Assembler()
+
+ self.assertEqual(packet._carried_packet, RawAssembler(data=b""))
+ self.assertEqual(packet._tracker, packet._carried_packet._tracker)
+ self.assertEqual(packet._ver, 6)
+ self.assertEqual(packet._dscp, 0)
+ self.assertEqual(packet._ecn, 0)
+ self.assertEqual(packet._flow, 0)
+ self.assertEqual(packet._hop, IP6_DEFAULT_HOP)
+ self.assertEqual(packet._src, Ip6Address(0))
+ self.assertEqual(packet._dst, Ip6Address(0))
+ self.assertEqual(packet._next, IP6_NEXT_RAW)
+ self.assertEqual(packet._dlen, 0)
+
+ def test_ip6_fpa__len(self):
+ """Test class __len__ operator"""
+
+ packet = Ip6Assembler()
+
+ self.assertEqual(len(packet), IP6_HEADER_LEN)
+
+ def test_ip6_fpa__len__data(self):
+ """Test class __len__ operator"""
+
+ packet = Ip6Assembler(
+ carried_packet=RawAssembler(data=b"0123456789ABCDEF"),
+ )
+
+ self.assertEqual(len(packet), IP6_HEADER_LEN + 16)
+
+ def test_ip6_fpa__str__(self):
+ """Test class __str__ operator"""
+
+ packet = Ip6Assembler(
+ src=Ip6Address("0:1:2:3:4:5:6:7"),
+ dst=Ip6Address("8:9:A:B:C:D:E:F"),
+ hop=32,
+ dscp=10,
+ ecn=2,
+ flow=12345678,
+ carried_packet=RawAssembler(data=b"0123456789ABCDEF"),
+ )
+
+ self.assertEqual(str(packet), "IPv6 0:1:2:3:4:5:6:7 > 8:9:a:b:c:d:e:f, next 255 (raw_data), flow 12345678, dlen 16, hop 32")
+
+ def test_ip6_fpa__tracker_getter(self):
+ """Test tracker getter"""
+
+ packet = Ip6Assembler()
+ self.assertTrue(repr(packet.tracker).startswith("Tracker(serial='<lr>TX"))
+
+ def test_ip6_fpa__dst_getter(self):
+ """Test dst getter"""
+
+ packet = Ip6Assembler(
+ dst=Ip6Address("8:9:A:B:C:D:E:F"),
+ )
+
+ self.assertEqual(packet.dst, Ip6Address("8:9:A:B:C:D:E:F"))
+
+ def test_ip6_fpa__src_getter(self):
+ """Test src getter"""
+
+ packet = Ip6Assembler(
+ src=Ip6Address("0:1:2:3:4:5:6:7"),
+ )
+
+ self.assertEqual(packet.src, Ip6Address("0:1:2:3:4:5:6:7"))
+
+ def test_ip6_fpa__dlen_getter(self):
+ """Test dlen getter"""
+
+ packet = Ip6Assembler(carried_packet=RawAssembler(data=b"0123456789ABCDEF"))
+
+ self.assertEqual(packet._dlen, 16)
+
+ def test_ip6_fpa__next_getter(self):
+ """Test next getter"""
+
+ packet = Ip6Assembler()
+
+ self.assertEqual(packet.next, IP6_NEXT_RAW)
+
+ def test_ip6_fpa__pshdr_sum(self):
+ """Test pshdr_sum getter"""
+
+ packet = Ip6Assembler(
+ src=Ip6Address("0:1:2:3:4:5:6:7"),
+ dst=Ip6Address("8:9:A:B:C:D:E:F"),
+ carried_packet=RawAssembler(data="0123456789ABCDEF"),
+ )
+
+ self.assertEqual(packet.pshdr_sum, 6755588421714211)
+
+ def test_ip6_fpa__assemble(self):
+ """Test assemble method"""
+
+ packet = Ip6Assembler(
+ src=Ip6Address("0:1:2:3:4:5:6:7"),
+ dst=Ip6Address("8:9:A:B:C:D:E:F"),
+ hop=32,
+ dscp=10,
+ ecn=2,
+ flow=12345678,
+ carried_packet=RawAssembler(data=b"0123456789ABCDEF"),
+ )
+
+ frame = memoryview(bytearray(IP6_HEADER_LEN + 16))
+ packet.assemble(frame)
+ self.assertEqual(bytes(frame),
+ b"`,aN\x00\x10\xff \x00\x00\x00\x01\x00\x02\x00\x03\x00\x04\x00\x05\x00\x06\x00"
+ b"\x07\x00\x08\x00\t\x00\n\x00\x0b\x00\x0c\x00\r\x00\x0e\x00\x0f0123456789ABCDEF"
+ )
Served by rngit 1.4.1 - Generated in 0.12s